package com.xy.webadmin.app.util;

import com.xy.webadmin.app.config.MqttSubscribeAdapterConfig;
import com.xy.webadmin.app.constant.MqttConstants;
import com.xy.webadmin.app.model.MqttBrokerConfig;
import com.xy.webadmin.app.service.MqttBrokerConfigService;
import com.xy.webadmin.app.service.MqttClientManager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;

/**
 * MQTT客户端工具类 - 基于Spring Integration MQTT
 *
 * @author xiaoyue
 * @date 2025-07-03
 */
@Slf4j
@Component
public class MqttClientUtil {

    @Autowired
    private MqttClientManager mqttClientManager;
    
    @Autowired
    private MqttBrokerConfigService mqttBrokerConfigService;
    
    @Autowired
    private MqttSubscribeAdapterConfig mqttSubscribeAdapterConfig;
    
    /**
     * 发布消息到MQTT主题
     *
     * @param companyId 公司ID
     * @param configId 配置ID
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级(0,1,2)
     * @param retained 是否保留消息
     * @return 成功标志
     */
    public boolean publishMessage(Long companyId, Long configId, String topic, String payload, int qos, boolean retained) {
        MessageHandler handler = mqttClientManager.getMqttOutboundHandler(companyId, configId);
        if (handler == null || !(handler instanceof MqttPahoMessageHandler)) {
            log.error("MQTT消息处理器未找到或类型错误, companyId: {}, configId: {}", companyId, configId);
            return false;
        }
        
        try {
            MqttPahoMessageHandler mqttHandler = (MqttPahoMessageHandler) handler;
            
            Message<String> message = MessageBuilder
                .withPayload(payload)
                .setHeader(MqttConstants.SpringIntegration.HEADER_TOPIC, topic)
                .setHeader(MqttConstants.SpringIntegration.HEADER_QOS, qos)
                .setHeader(MqttConstants.SpringIntegration.HEADER_RETAINED, retained)
                .build();
            
            mqttHandler.handleMessage(message);
            log.debug("已发布消息到主题: {}, configId: {}", topic, configId);
            return true;
        } catch (Exception e) {
            log.error("发布MQTT消息失败, topic: {}, error: {}", topic, e.getMessage(), e);
            return false;
        }
    }
    
    /**
     * 使用默认QoS发布消息
     *
     * @param companyId 公司ID
     * @param configId 配置ID
     * @param topic 主题
     * @param payload 消息内容
     * @return 成功标志
     */
    public boolean publishMessage(Long companyId, Long configId, String topic, String payload) {
        return publishMessage(companyId, configId, topic, payload, MqttConstants.QosLevel.AT_MOST_ONCE, false);
    }
    
    /**
     * 订阅MQTT主题
     * 在Spring Integration中，订阅是通过配置适配器完成的
     *
     * @param configId 配置ID
     * @param topics 主题数组
     * @param qos QoS级别数组
     * @return 成功标志
     */
    public boolean subscribe(Long configId, String[] topics, int[] qos) {
        try {
            MqttBrokerConfig config = mqttBrokerConfigService.getById(configId);
            if (config == null) {
                log.error("MQTT配置不存在, configId: {}", configId);
                return false;
            }
            
            mqttSubscribeAdapterConfig.createSubscribeAdapter(config, topics, qos);
            return true;
        } catch (Exception e) {
            log.error("创建MQTT订阅失败, configId: {}, error: {}", configId, e.getMessage(), e);
            return false;
        }
    }
    
    /**
     * 订阅单个主题
     *
     * @param configId 配置ID
     * @param topic 主题
     * @param qos QoS级别
     * @return 成功标志
     */
    public boolean subscribe(Long configId, String topic, int qos) {
        return subscribe(configId, new String[]{topic}, new int[]{qos});
    }
} 